package cn.com.sparkle.paxos.net;

import java.net.InetSocketAddress;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.log4j.Logger;

import cn.com.sparkle.paxos.NodeState;
import cn.com.sparkle.paxos.deamon.ReConnectThread;
import cn.com.sparkle.paxos.deamon.ReConnectThread.ReConnectMethod;
import cn.com.sparkle.paxos.event.EventsManager;
import cn.com.sparkle.paxos.event.NodeStateChangeEvent;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.MessagePackage;
import cn.com.sparkle.raptor.core.handler.IoHandler;
import cn.com.sparkle.raptor.core.protocol.MultiThreadProtecolHandler.ProtocolHandlerIoSession;
import cn.com.sparkle.raptor.core.protocol.ProtocolHandler;
import cn.com.sparkle.raptor.core.transport.socket.nio.IoSession;
import cn.com.sparkle.raptor.core.transport.socket.nio.NioSocketClient;

public class ClientNetHandler implements ProtocolHandler {
	private final static Logger logger = Logger
			.getLogger(ClientNetHandler.class);

	private final NioSocketClient client;
	private IoHandler handler;
	private ReentrantLock lock = new ReentrantLock();
	private EventsManager eventsManager;
	private ReConnectThread reConnectThread;

	public ClientNetHandler(NioSocketClient client,
			EventsManager eventsManager, ReConnectThread reConnectThread) {
		this.client = client;
		this.eventsManager = eventsManager;
		this.reConnectThread = reConnectThread;
	}

	public void setHandler(IoHandler handler) {
		this.handler = handler;
	}

	@Override
	public void onOneThreadSessionOpen(ProtocolHandlerIoSession session) {

		logger.debug("open connected session" + session.toString());
		NodeState nodeState = (NodeState) session.customAttachment;
		nodeState.setConnected(true);
		NetNode nnode = new NetNode(session, nodeState);
		try {
			lock.lock();
			NodeStateChangeEvent.doOpenConnectEvent(eventsManager, nnode);
		} finally {
			lock.unlock();
		}

		session.customAttachment = nnode;
	}

	@Override
	public void onOneThreadSessionClose(ProtocolHandlerIoSession attachment) {
		NetNode nnode = (NetNode) attachment.customAttachment;
		nnode.nodeState.setConnected(false);

		try {
			lock.lock();
			NodeStateChangeEvent.doLoseConnectEvent(eventsManager, nnode);
		} finally {
			lock.unlock();
		}
		nnode.close();
		reConnect(nnode.nodeState);
	}

	@Override
	public void onOneThreadCatchException(IoSession session,
			ProtocolHandlerIoSession attachment, Throwable e) {
		e.printStackTrace();
		if (attachment == null) {
			NodeState ns = (NodeState) session.attachment();
			reConnect(ns);
		}
	}

	@Override
	public void onOneThreadMessageRecieved(Object receiveObject,
			ProtocolHandlerIoSession attachment) {
		NetNode nnode = (NetNode) attachment.customAttachment;
		MessagePackage mp = (MessagePackage) receiveObject;
		nnode.recieve(mp);
	}

	@Override
	public void onOneThreadMessageSent(ProtocolHandlerIoSession attachment,
			int sendSize) {
	}

	private final class ReConnect implements ReConnectMethod {
		private final Logger log = Logger.getLogger(ReConnect.class);

		@Override
		public void reConnect(Object value) {
			NodeState ns = (NodeState) value;
			String address = ns.getAddress();
			String[] a = address.split(":");
			try {
				log.debug("reConnect " + address);
				client.connect(
						new InetSocketAddress(a[0], Integer.parseInt(a[1])),
						handler, ns);
			} catch (Exception e) {
				reConnectThread.add(ns, this, 2000);
			}
		}

	}

	private final ReConnect method = new ReConnect();

	private void reConnect(NodeState ns) {
		// After sleep 5 seconds,retry to connect.
		reConnectThread.add(ns, method, 5000);
	}

}
